package com.alibaba.datax.plugin.reader.aliyunslsreader;

import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
//import com.alibaba.datax.plugin.reader.aliyunslsreader.util.LogHubReaderUtil;
import com.alibaba.datax.plugin.reader.aliyunslsreader.util.LogHubReaderUtil;
import com.alibaba.datax.plugin.unstructuredstorage.reader.ColumnEntry;
import com.alibaba.datax.plugin.unstructuredstorage.reader.Key;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderUtil;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.log.common.FastLog;
import com.aliyun.openservices.log.common.FastLogContent;
import com.aliyun.openservices.log.common.FastLogGroup;
import com.aliyun.openservices.log.common.LogGroupData;
import com.aliyun.openservices.loghub.client.ILogHubCheckPointTracker;
import com.aliyun.openservices.loghub.client.exceptions.LogHubCheckPointException;
import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;

import java.util.List;

public class LogHubProcessor implements ILogHubProcessor {
    private int shardId;
    public void initialize(int shardId) {
        this.shardId = shardId;
    }
    private RecordSender recordSender;
    public void setRecordSender(RecordSender recordSender) {
        this.recordSender = recordSender;
    }

    private Configuration configuration;
    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    private TaskPluginCollector taskPluginCollector;
    public void setTaskPluginCollector(TaskPluginCollector taskPluginCollector) {
        this.taskPluginCollector = taskPluginCollector;
    }

    // 消费数据的主逻辑，消费时的所有异常都需要处理，不能直接抛出。
    public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
        // 打印已获取的数据。
        for (LogGroupData logGroup : logGroups) {
            FastLogGroup flg = logGroup.GetFastLogGroup();
            for (int lIdx = 0; lIdx < flg.getLogsCount(); ++lIdx) {
                FastLog log = flg.getLogs(lIdx);

                JSONObject oneRecordJson = new JSONObject();
                for (int cIdx = 0; cIdx < log.getContentsCount(); ++cIdx) {
                    FastLogContent content = log.getContents(cIdx);
                    oneRecordJson.put(content.getKey(), content.getValue());
                    System.out.print(content.getKey() + ":" + content.getValue() + "\n");
                }

                Record record = LogHubReaderUtil.getRecord(this.recordSender, this.configuration,
                        oneRecordJson, taskPluginCollector);

                recordSender.sendToWriter(record);
            }
        }

        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
        return null;
    }

    // 当Worker退出时，会调用该函数，您可以在此处执行清理工作。
    public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
        //将Checkpoint立即保存到服务端。
        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
    }
}